package com.yangzb;

import cn.hutool.core.io.resource.ClassPathResource;
import cn.hutool.core.lang.UUID;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.google.common.base.Stopwatch;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.LocatedFileStatus;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: yangzb
 * @date 2023/1/3 11:22 AM
 **/
@Slf4j
public class HdfsToClickHouse {
    protected ThreadPoolExecutor importThreadPoolExecutor = (ThreadPoolExecutor) ThreadUtil.newExecutor(1, 100, Integer.MAX_VALUE);
    protected ThreadPoolExecutor downloadThreadPoolExecutor = (ThreadPoolExecutor) ThreadUtil.newExecutor(1, 100, Integer.MAX_VALUE);
    protected HdfsConfig hdfsConfig;
    protected ClickHouseConfig clickHouseConfig;
    protected Set<TaskInfo> results = new HashSet<>();
    protected Set<Future<TaskInfo>> futures = new HashSet<>();


    public HdfsToClickHouse(@NonNull HdfsConfig hdfsConfig, @NonNull ClickHouseConfig clickHouseConfig) {
        this.hdfsConfig = hdfsConfig;
        this.clickHouseConfig = clickHouseConfig;
    }

    public Result run() throws Exception {
        Stopwatch stopwatch = Stopwatch.createStarted();
        Set<LocatedFileStatus> locatedFileStatuses = HdfsUtil.listFiles(hdfsConfig.getHdfsPath(), true);
        if (locatedFileStatuses == null || locatedFileStatuses.size() < 1) {
            throw new RuntimeException(StrUtil.format("hdfs目录：{}，无需要处理的文件", hdfsConfig.getHdfsPath()));
        }
        String folder = UUID.fastUUID().toString(true);
        Iterator<LocatedFileStatus> iterator = locatedFileStatuses.iterator();
        if(StrUtil.isBlank(hdfsConfig.getLocalPath())){
            String path = System.getProperty("user.dir");
            hdfsConfig.setLocalPath(path+"/download");
        }
        while (iterator.hasNext()) {
            LocatedFileStatus next = iterator.next();
            String localFile = hdfsConfig.getLocalPath() + "/" + folder + "/" + next.getPath().getName();
            TaskInfo taskInfo = new TaskInfo(next.getPath().toUri().getPath(),localFile);
            futures.add(downloadThreadPoolExecutor.submit(new HdfsDownloader(hdfsConfig, taskInfo, taskInfo1 -> {
                futures.add(importThreadPoolExecutor.submit(new ClickhouseWriter(taskInfo1, clickHouseConfig)));
            })));
        }

        while (true) {
            int importActiveCount = importThreadPoolExecutor.getActiveCount();
            int downloadActiveCount = downloadThreadPoolExecutor.getActiveCount();

            log.info("计划任务数：{}，下载完成任务：{}，导入完成任务：{}",locatedFileStatuses.size(),downloadThreadPoolExecutor.getTaskCount(),importThreadPoolExecutor.getTaskCount());
            if (importActiveCount == 0 && downloadActiveCount == 0) {
                break;
            }
            try {
                TimeUnit.SECONDS.sleep(30);
            } catch (Exception e) {
                log.error("", e);
            }
        }

        log.info("关闭线程池:{}",stopwatch.stop().toString());
        importThreadPoolExecutor.shutdown();
        downloadThreadPoolExecutor.shutdown();
        log.info("任务处理完毕");

        Iterator<Future<TaskInfo>> futureIterator = futures.iterator();
        while (futureIterator.hasNext()) {
            Future<TaskInfo> next = futureIterator.next();
            results.add(next.get());
        }
        Result result = new Result();
        for (TaskInfo taskInfo : results) {
            if(taskInfo.isSuccess()){
                result.getSuccessList().add(taskInfo);
            }else {
                result.getFailedList().add(taskInfo);
            }
        }
        result.setTaskCount(locatedFileStatuses.size());
        return result;
    }
}
